package service

import (
	"context"
	"gitee.com/man0sions/go-cron/common"
	"github.com/coreos/etcd/clientv3"
	"github.com/coreos/etcd/mvcc/mvccpb"
	"log"
	"strings"
)

type JobMgr struct {
}

var (
	GJobMgr *JobMgr
)

func (jm *JobMgr) WatchJobs() {
	var (
		getRes *clientv3.GetResponse
		job    *common.Job
		err    error
	)
	if getRes, err = common.GEtcd.Kv.Get(context.TODO(), common.ETCD_JOBS_DIR, clientv3.WithPrefix()); err != nil {
		return
	}

	//加入默认的job
	for _, v := range getRes.Kvs {
		if job, err = common.UnPackJob(v.Value); err != nil {
			common.Warn(nil, err)
			continue
		}
		job.Type = common.JOB_TYPE_PUT
		log.Printf("%+v", job)
		Gscheduler.AddTask(job)

	}

	go func() {
		ch := common.GEtcd.Watcher.Watch(context.TODO(), common.ETCD_JOBS_DIR, clientv3.WithPrefix(), clientv3.WithRev(getRes.Header.Revision+1))
		for i := range ch {
			for _, event := range i.Events {

				switch event.Type {
				case mvccpb.PUT:
					if job, err = common.UnPackJob(event.Kv.Value); err != nil {
						common.Warn(nil, err)
						continue
					}
					job.Type = common.JOB_TYPE_PUT
					Gscheduler.AddTask(job)

				case mvccpb.DELETE:
					name := strings.Replace(string(event.Kv.Key), common.ETCD_JOBS_DIR, "", -1)
					job = &common.Job{
						Name: name,
						Type: common.JOB_TYPE_DEL,
					}
					Gscheduler.AddTask(job)

				}

			}

		}
	}()

	return
}

func (jm *JobMgr) WatchKill() {
	go func() {
		ch := common.GEtcd.Watcher.Watch(context.TODO(), common.ETCD_KILL_DIR, clientv3.WithPrefix())
		for i := range ch {
			for _, event := range i.Events {
				switch event.Type {
				case mvccpb.PUT:
					job := &common.Job{
						Name: strings.TrimPrefix(string(event.Kv.Key), common.ETCD_KILL_DIR),
						Type: common.JOB_TYPE_KILL,
					}
					Gscheduler.AddTask(job)

				}
			}
		}
	}()

}

func InitJobMgr() (err error) {

	GJobMgr = &JobMgr{}

	go GJobMgr.WatchJobs()
	go GJobMgr.WatchKill()

	return

}
